# Week 3 | 第4课：Web UI 集成

**课程编号**: 3.4
**时长**: 30 分钟
**前置**: 3.3 实战 —— 多步骤决策流程

---

## 学习目标

- 使用 Gradio 为 LangGraph Agent 搭建 Web UI
- 实现聊天界面：输入框 + 提交按钮 + 对话历史
- 将 Agent 的执行过程（状态变化）实时展示在 UI 上
- 了解 Gradio Share、Hugging Face Spaces 等部署选项

---

## 1. 为什么用 Gradio？

在 Python 生态中，给 AI 应用加 Web UI 有几个选择：

| 方案 | 安装难度 | 代码量 | 自定义程度 | 适合场景 |
|------|---------|--------|-----------|---------|
| **Gradio** | `pip install gradio` | 少 | 中 | 快速原型、Demo、分享 |
| Streamlit | `pip install streamlit` | 中 | 中高 | 数据应用、仪表盘 |
| FastAPI + HTML | 多 | 多 | 高 | 生产环境、前后端分离 |
| Flask + HTML | 中 | 多 | 高 | 传统 Web 应用 |

对于 Agent 应用的原型和演示，**Gradio 是最快的选择**：几十行代码就能得到一个带聊天界面的 Web App。

---

## 2. 环境准备

```bash
pip install gradio langgraph langchain-openai
```

---

## 3. 最简版：给 Agent 加聊天界面

先从一个最简单的版本开始 —— 只连接 LLM，不用 LangGraph。

```python
"""
lesson3_4_basic_chat.py
最简 Gradio 聊天界面：连接 LLM
"""

import gradio as gr
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chat(message: str, history: list) -> list:
    """
    Gradio ChatInterface 的回调函数
    message: 用户刚输入的消息（字符串）
    history: 对话历史（列表 of [user_msg, bot_msg]）
    返回值: 更新后的对话历史
    """
    # 将 Gradio 的 history 格式转换为 LangChain 的 messages 格式
    messages = []
    for user_msg, bot_msg in history:
        messages.append(HumanMessage(content=user_msg))
        messages.append(AIMessage(content=bot_msg))

    # 加上当前消息
    messages.append(HumanMessage(content=message))

    # 调用 LLM
    response = llm.invoke(messages)

    # 返回新的对话历史（Gradio 会自动更新 UI）
    return history + [[message, response.content]]


# 创建聊天界面
demo = gr.ChatInterface(
    fn=chat,
    title="AI 客服助手",
    description="输入你的问题，AI 客服会为你解答",
    examples=["账单问题", "技术故障", "办公时间"],
)

if __name__ == "__main__":
    demo.launch()
    # 运行后访问: http://localhost:7860
```

**运行方式：**
```bash
python lesson3_4_basic_chat.py
```

打开浏览器访问 `http://localhost:7860`，你会看到一个完整的聊天界面。

---

## 4. 完整版：连接 LangGraph Agent

现在将 3.3 课中的客服分类系统连接到 Gradio UI。

```python
"""
lesson3_4_agent_ui.py
将 LangGraph 客服分类系统连接到 Gradio Web UI
功能：聊天界面 + 执行路径可视化
"""

import gradio as gr
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# ========== LangGraph 部分 ==========

# 初始化 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class TriageState(TypedDict):
    messages: Annotated[List, add_messages]
    intent: str
    reply: str
    satisfied: bool
    retry_count: int
    max_retries: int
    execution_path: list  # 记录执行路径用于 UI 展示

# ----- Nodes -----
def classify_intent(state: TriageState) -> dict:
    last_msg = state["messages"][-1]
    question = last_msg.content if hasattr(last_msg, 'content') else str(last_msg)

    system_prompt = """你是一个客服意图分类器。分类：
- billing: 账单、付款、发票、费用、退款
- tech: 技术故障、功能使用、API、bug
- support: 其他一般问题
只返回类别名称（billing/tech/support）。"""

    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=question)
    ])
    intent = response.content.strip().lower()

    path = state.get("execution_path", [])
    return {"intent": intent, "execution_path": path + [f"分类({intent})"]}

def handle_billing(state: TriageState) -> dict:
    prompt = f"你是账单专家。请回答：{state['messages'][-1].content}"
    response = llm.invoke([HumanMessage(content=prompt)])
    path = state.get("execution_path", [])
    return {"reply": response.content, "execution_path": path + ["账单处理"]}

def handle_tech(state: TriageState) -> dict:
    prompt = f"你是技术支持专家。请回答：{state['messages'][-1].content}"
    response = llm.invoke([HumanMessage(content=prompt)])
    path = state.get("execution_path", [])
    return {"reply": response.content, "execution_path": path + ["技术处理"]}

def handle_support(state: TriageState) -> dict:
    prompt = f"你是友好客服。请回答：{state['messages'][-1].content}"
    response = llm.invoke([HumanMessage(content=prompt)])
    path = state.get("execution_path", [])
    return {"reply": response.content, "execution_path": path + ["一般处理"]}

def check_satisfaction(state: TriageState) -> dict:
    reply = state["reply"]
    last_user = ""
    for msg in reversed(state["messages"]):
        if isinstance(msg, HumanMessage):
            last_user = msg.content
            break

    negative = ["不行", "没解决", "还是不对", "不满意", "不对", "错误"]
    satisfied = not any(w in last_user for w in negative)

    path = state.get("execution_path", [])
    return {
        "satisfied": satisfied,
        "messages": [AIMessage(content=reply)],
        "execution_path": path + [f"满意度({'满意' if satisfied else '不满意'})"],
    }

def handle_retry(state: TriageState) -> dict:
    retry_count = state["retry_count"] + 1
    path = state.get("execution_path", [])
    return {
        "retry_count": retry_count,
        "messages": [HumanMessage(content="用户不满意，请重新分析并提供更好的方案。")],
        "execution_path": path + [f"重试({retry_count})"],
    }

# ----- Routing -----
def route_by_intent(state): return state["intent"]

def route_by_satisfaction(state):
    if state["satisfied"]: return "end"
    if state["retry_count"] >= state.get("max_retries", 3): return "max_retries"
    return "retry"

# ----- Build Graph -----
builder = StateGraph(TriageState)
builder.add_node("classify", classify_intent)
builder.add_node("billing", handle_billing)
builder.add_node("tech", handle_tech)
builder.add_node("support", handle_support)
builder.add_node("check", check_satisfaction)
builder.add_node("retry", handle_retry)

builder.set_entry_point("classify")
builder.add_conditional_edges("classify", route_by_intent, {
    "billing": "billing", "tech": "tech", "support": "support"
})
builder.add_edge("billing", "check")
builder.add_edge("tech", "check")
builder.add_edge("support", "check")
builder.add_conditional_edges("check", route_by_satisfaction, {
    "end": END, "retry": "retry", "max_retries": END
})
builder.add_edge("retry", "classify")

agent = builder.compile()

# ========== Gradio UI 部分 ==========

def process_message(message: str, history: list) -> str:
    """
    处理用户消息，调用 LangGraph Agent
    返回值: Agent 的回复文本
    """
    initial_state = {
        "messages": [HumanMessage(content=message)],
        "intent": "",
        "reply": "",
        "satisfied": False,
        "retry_count": 0,
        "max_retries": 3,
        "execution_path": ["Start"],
    }

    result = agent.invoke(initial_state)

    # 获取执行路径，方便用户看到处理过程
    path = " → ".join(result.get("execution_path", []))
    print(f"执行路径: {path}")

    return result["reply"]


def create_ui():
    """创建 Gradio 聊天界面"""
    with gr.Blocks(title="AI 客服系统", theme=gr.themes.Soft()) as demo:
        gr.Markdown("# AI 智能客服系统")
        gr.Markdown("基于 LangGraph 状态机，支持意图分类、智能路由和满意度循环")

        # 聊天组件
        chatbot = gr.Chatbot(
            label="对话",
            height=400,
            type="tuples",  # [(user, bot), ...] 格式
        )

        with gr.Row():
            msg_input = gr.Textbox(
                label="输入问题",
                placeholder="请输入你的问题...",
                scale=4,
            )
            submit_btn = gr.Button("发送", variant="primary", scale=1)

        # 执行路径展示
        path_display = gr.Markdown("**执行路径**: 等待输入...")

        # 示例问题
        gr.Examples(
            examples=[
                ["我这个月账单多了50块钱？"],
                ["你们的 App 一直闪退"],
                ["请问办公时间是什么？"],
            ],
            inputs=msg_input,
        )

        def respond(message, chat_history):
            """点击发送时的处理"""
            if not message or not message.strip():
                return "", chat_history + [["", "请输入你的问题"]], "**执行路径**: 无输入"

            # 调用 Agent
            reply = process_message(message, chat_history)

            chat_history.append([message, reply])

            # 构建执行路径展示（这里简化处理）
            path_text = f"**处理流程**: Start → 意图分类 → 路由处理 → 满意度检查 → 完成"

            return "", chat_history, path_text

        # 绑定事件
        submit_btn.click(
            respond,
            inputs=[msg_input, chatbot],
            outputs=[msg_input, chatbot, path_display],
        )
        msg_input.submit(
            respond,
            inputs=[msg_input, chatbot],
            outputs=[msg_input, chatbot, path_display],
        )

    return demo


if __name__ == "__main__":
    app = create_ui()
    app.launch()
    # 访问 http://localhost:7860
```

---

## 5. 流式输出版本

如果希望 AI 回复时逐字输出（类似 ChatGPT 的效果），可以用 Gradio 的 generator 模式。

```python
"""
lesson3_4_streaming_ui.py
流式输出的 Gradio UI
"""

import gradio as gr
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)

def stream_chat(message: str, history: list):
    """
    Generator 函数：逐字 yield 回复
    """
    response = ""
    for chunk in llm.stream([HumanMessage(content=message)]):
        response += chunk.content
        # 每次 yield 都会更新 UI 中的输出
        yield response

demo = gr.ChatInterface(
    fn=stream_chat,
    title="AI 客服（流式）",
    description="AI 会逐字输出回复",
)

if __name__ == "__main__":
    demo.launch()
```

**关键点**：当 `fn` 是一个 generator（使用 `yield`）时，Gradio 会自动实现流式输出效果。每次 `yield` 的值会替换 UI 中正在显示的文本。

---

## 6. 部署选项

### 6.1 Gradio Share（最快分享方式）

```python
if __name__ == "__main__":
    app.launch(share=True)
    # 会生成一个临时的公网 URL，如：
    # https://xxx-xxx-xxx.gradio.live
    # 有效期 72 小时
```

一行代码即可分享给别人，适合演示和测试。

### 6.2 Hugging Face Spaces（免费托管）

```yaml
# Hugging Face Spaces 只需一个 app.py 和一个 requirements.txt
# requirements.txt:
gradio
langgraph
langchain-openai
```

步骤：
1. 在 [huggingface.co/spaces](https://huggingface.co/spaces) 创建新 Space
2. 选择 SDK 为 Gradio
3. 上传 `app.py`（你的 Gradio 代码）和 `requirements.txt`
4. 在 Space 设置中添加 `OPENAI_API_KEY` 环境变量
5. 自动部署，获得永久 URL

### 6.3 自己的服务器

```bash
# 安装生产级服务器
pip install uvicorn

# 用 Gradio 的 API 模式
if __name__ == "__main__":
    app.launch(
        server_name="0.0.0.0",  # 允许外部访问
        server_port=7860,
        share=False,
    )
```

生产环境建议：
- 使用 `nginx` 做反向代理
- 配置 HTTPS
- 设置请求限流
- 管理 API Key（不要硬编码在代码里）

---

## 7. 动手练习

### 练习 1：给 UI 添加状态监控面板

在 Gradio UI 中添加一个实时状态面板，显示 Agent 当前所在的节点：

```python
# 在 Blocks 中添加一个状态显示区域
status_box = gr.Textbox(label="当前状态", value="等待输入", interactive=False)

# 在 respond 函数中，使用 LangGraph 的 stream 方法获取中间状态
for event in agent.stream(initial_state):
    node_name = list(event.keys())[0]
    status_text = f"正在执行: {node_name}"
    # 更新状态显示（需要使用 Gradio 的事件更新机制）
```

提示：使用 `agent.stream()` 而非 `agent.invoke()` 可以获取每一步的执行状态。

### 练习 2：添加对话历史记录

在 UI 中添加一个"清除对话"按钮和对话历史记录功能：

```python
clear_btn = gr.Button("清除对话")

def clear_chat():
    return [], "**执行路径**: 已清除"

clear_btn.click(clear_chat, outputs=[chatbot, path_display])
```

### 练习 3：多语言支持

给 UI 添加一个语言选择下拉菜单，支持中文和英文：

```python
language = gr.Dropdown(
    choices=["中文", "English"],
    value="中文",
    label="界面语言",
)
```

根据选择的语言动态更新 system prompt。

---

## 8. 小结

本课要点：

- **Gradio** 是 Python AI 应用最快的 UI 方案，几十行代码即可得到聊天界面
- `gr.ChatInterface` 适合简单场景，`gr.Blocks` 提供更多自定义能力
- 使用 generator + `yield` 可以实现流式输出效果
- 使用 `agent.stream()` 可以获取每一步的执行状态，实现实时进度展示
- 部署选项：`share=True` 临时分享、Hugging Face Spaces 免费托管、自建服务器生产部署
- Gradio 适合原型和演示，生产环境可能需要 FastAPI + 前端框架

**Week 3 课程总结**:
- 3.1 理解了为什么需要状态机（可控、可调试、可可视化）
- 3.2 掌握了 LangGraph 的核心概念（State、Node、Edge、Conditional Edge）
- 3.3 构建了完整的多步骤决策系统（分类 → 路由 → 处理 → 满意度 → 循环）
- 3.4 给 Agent 加上了 Web UI（Gradio），可以演示和分享

**Week 4 预告**: MCP（Model Context Protocol）—— 让 Agent 连接外部工具和数据源。